package com.yangzb.flink.stream.sample;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.io.resource.ClassPathResource;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

@Slf4j
public class Main {
    private static Properties getProperties() throws IOException {
        ClassPathResource classPathResource = new ClassPathResource("application.properties");
        return ParameterTool.fromPropertiesFile(classPathResource.getAbsolutePath()).getProperties();
    }

    public static void main(String[] args) throws Exception {
        Properties config = getProperties();
        log.info(JSONUtil.toJsonPrettyStr(config));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.milliseconds(30000)));
        env.disableOperatorChaining();
        env.setParallelism(1);
        // 每 1000ms 开始一次 checkpoint
        env.enableCheckpointing(1_000 * 60 * 5, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.enableUnalignedCheckpoints();
        // 高级选项：

        // 设置模式为精确一次 (这是默认值)
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 确认 checkpoints 之间的时间会进行 500 ms
        checkpointConfig.setMinPauseBetweenCheckpoints(500);

        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
        checkpointConfig.setCheckpointTimeout(1000 * 60 * 10);

        // 同一时间只允许一个 checkpoint 进行
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        checkpointConfig.setTolerableCheckpointFailureNumber(5);

        // 开启在 job 中止后仍然保留的 externalized checkpoints
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setStateBackend(new RocksDBStateBackend(config.getProperty("flink.checkpoint.data.uri"), true));

        flatMap(env);
        env.execute("test");

    }


    private static void flatMap(StreamExecutionEnvironment env) {
        DataStreamSource<String> streamSource = env.fromCollection(getData());
        streamSource.print();
        log.error("===============================");
        SingleOutputStreamOperator<Integer> outputStreamOperator = streamSource.flatMap(new FlatMapFunction<String, Integer>() {
            @Override
            public void flatMap(String s, Collector<Integer> collector) throws Exception {
                char[] chars = s.toCharArray();
                for (int i = 0; i < chars.length; i++) {
                    collector.collect(Convert.toInt(chars[i]));
                }
            }
        });
        outputStreamOperator.print();

    }

    private static List<String> getData() {
        List<String> datas = new ArrayList<>();
        for (int i = 0; i < 1; i++) {
            datas.add(Convert.toStr(RandomUtil.randomInt(0, 20)));
        }
        return datas;
    }
}
